// testqueue.cpp : 此文件包含 "main" 函数。程序执行将在此处开始并结束。
//
#include "public.h"
#include "typedefine.h"
#include "BF_Queue.h"
#include "BF_Tools.h"
#include "BF_Thread.h"
#include "MdbBase.h"

#include <iostream>

class CInt64
{
public:
    std::set<UINT64_> m_intdata;
    std::set<UINT64_>::iterator m_current;
    std::set<UINT64_>::iterator m_end;
    std::set<UINT64_>::iterator m_begin;

public:
    CInt64()
    {
    }
    ~CInt64()
    {
        Clear();
    }
    void Add(UINT64_ id)
    {
        m_intdata.insert(id);
    }
    bool First(UINT64_ &id)
    {
        m_current = m_intdata.begin();
        m_end = m_intdata.end();
        if (m_current == m_end)
        {
            return false;
        }
        else
        {
            id = *m_current;
            return true;
        }
    }
    bool Next(UINT64_ &id)
    {
        m_current++;
        if (m_current == m_end)
        {
            return false;
        }
        else
        {
            id = *m_current;
            return true;
        }
    }
    bool End(UINT64_ &id)
    {
        m_current = m_intdata.end();
        m_begin = m_intdata.begin();
        if (m_current == m_end)
        {
            return false;
        }
        else
        {
            m_current--;
            id = *m_current;
            return true;
        }
    }
    bool Last(UINT64_ &id)
    {
        if (m_current == m_begin)
        {
            return false;
        }
        else
        {
            m_current--;
            id = *m_current;
            return true;
        }
    }
    bool Find(UINT64_ id)
    {
        std::set<UINT64_>::iterator p;
        p = m_intdata.find(id);
        if (p == m_intdata.end())
        {
            return false;
        }
        else
        {
            return true;
        }
    }
    int Delete(UINT64_ id)
    {
        return m_intdata.erase(id);
    }
    void Clear()
    {
        m_intdata.clear();
    }
    int Size()
    {
        return m_intdata.size();
    }
};

// 多个64位无符号整型字段的key，不可重复
template <int count>
class CPkeyUint64_TEST
{
public:
    struct Int64_
    {
        UINT64_ k[count]; // 多个整型索引
        UINT64_ id;       // 对应表的ID，相当于rowid
    };
    struct Int64_lt
    {
        bool operator()(const Int64_ f1, const Int64_ f2) const
        {
            INT64_ r = 0;
            int i;

            for (i = 0; i < count; i++)
            {
                if (f1.k[i] == f2.k[i])
                {
                    continue;
                }
                else
                {
                    if (f1.k[i] > f2.k[i])
                    {
                        r = 1;
                    }
                    else
                    {
                        r = -1;
                    }
                    break;
                }
                // r=f1.k[i]-f2.k[i];
                // if (r != 0)
                //{
                //	break;
                // }
            }
            return r < 0;
        }
    };

    CUniqueIndexBase<Int64_, Int64_lt> m_key; // key数据，不可重复

public:
    CPkeyUint64_TEST()
    {
    }
    ~CPkeyUint64_TEST()
    {
    }
    // 增加一条索引:rowid,索引...
    void Add(UINT64_ id, UINT64_ ind, ...)
    {
        Int64_ a_;
        va_list intp;
        int i;

        a_.id = id;
        a_.k[0] = ind;
        va_start(intp, ind);
        for (i = 0; i < count - 1; i++)
        {
            a_.k[i + 1] = va_arg(intp, UINT64_);
        }
        va_end(intp);
        m_key.Add(a_);
        return;
    }
    // 查找指定key的rowid  删除时使用
    bool Select(CInt64 &iset, UINT64_ k, ...)
    {
        Int64_ a_;
        bool bret_;
        int i;
        va_list intp;

        iset.Clear();
        a_.k[0] = k;
        va_start(intp, k);
        for (i = 0; i < count - 1; i++)
        {
            a_.k[i + 1] = va_arg(intp, UINT64_);
        }
        va_end(intp);
        bret_ = m_key.Find(a_);
        if (bret_)
        {
            iset.Add(a_.id);
        }
        return bret_;
    }
    // 查找指定key的rowid
    bool Select(UINT64_ &rid, UINT64_ k, ...)
    {
        Int64_ a_;
        bool bret_;
        int i;
        va_list intp;

        a_.k[0] = k;
        va_start(intp, k);
        for (i = 0; i < count - 1; i++)
        {
            a_.k[i + 1] = va_arg(intp, UINT64_);
        }
        va_end(intp);
        bret_ = m_key.Find(a_);
        if (bret_)
        {
            rid = a_.id;
        }
        return bret_;
    }
    // 查找指定key的数据是否存在
    bool Find(UINT64_ k, ...)
    {
        Int64_ a_;
        bool bret_;
        int i;
        va_list intp;

        a_.k[0] = k;
        va_start(intp, k);
        for (i = 0; i < count - 1; i++)
        {
            a_.k[i + 1] = va_arg(intp, UINT64_);
        }
        va_end(intp);
        bret_ = m_key.Find(a_);
        return bret_;
    }
    // 删除指定key
    int Delete(UINT64_ k, ...)
    {
        Int64_ a_;
        va_list intp;
        int i;

        a_.k[0] = k;
        va_start(intp, k);
        for (i = 0; i < count - 1; i++)
        {
            a_.k[i + 1] = va_arg(intp, UINT64_);
        }
        va_end(intp);
        return m_key.Erase(a_);
    }
    // 扫描全表，删除rowid
    void Delete(CInt64 iset)
    {
        Int64_ a_;
        bool bret_;
        bret_ = m_key.First(a_);
        while (bret_)
        {
            if (iset.Find(a_.id))
            {
                m_key.Erase(a_);
            }
            bret_ = m_key.Next(a_);
        }
    }
    bool First(UINT64_ &id)
    {
        Int64_ aa;
        if (!m_key.First(aa))
        {
            return false;
        }
        id = aa.id;
        return true;
    }
    bool Next(UINT64_ &id)
    {
        Int64_ aa;
        if (!m_key.Next(aa))
        {
            return false;
        }
        id = aa.id;
        return true;
    }
    bool End(UINT64_ &rid)
    {
        Int64_ aa;
        if (!m_key.End(aa))
        {
            return false;
        }
        rid = aa.id;
        return true;
    }
    bool Last(UINT64_ &id)
    {
        Int64_ aa;
        if (!m_key.Last(aa))
        {
            return false;
        }
        id = aa.id;
        return true;
    }
    void Clear()
    {
        m_key.Clear();
    }
    UINT64_ Size()
    {
        return m_key.Size();
    }
};

typedef struct _NOLOCK_DATA
{
    UINT64_ data_id;
    UINT64_ push_us;
    int push_ns;
} S_NOLOCK_DATA;
CBF_QueueNoLock<S_NOLOCK_DATA, 20000> g_QueueNoLock;
int g_nExitPush;
int g_nPushSleepTimeus;

class CTbl_PopData
{
protected:
    CMemTableNew<UINT64_> m_table;
    CPkeyUint64_TEST<1> m_pkey; // 交易账号
public:
    CTbl_PopData()
    {
    }
    virtual ~CTbl_PopData()
    {
    }
    void Clear()
    {
        m_table.Clear();
        m_pkey.Clear();
    }
    bool Add(S_NOLOCK_DATA info)
    {
        if (m_pkey.Find(info.data_id))
        {
            return false;
        }
        UINT64_ rid = m_table.Add(info.data_id);
        m_pkey.Add(rid, m_table.m_table[rid]);
        return true;
    }
    int Size()
    {
        return m_pkey.Size();
    }
    bool Select(UINT64_ id)
    {
        return m_pkey.Find(id);
    }
    UINT64_ *First()
    {
        UINT64_ rid;
        bool bret = m_pkey.First(rid);
        if (!bret)
        {
            return NULL;
        }
        return &m_table.m_table[rid];
    }
    UINT64_ *Next()
    {
        UINT64_ rid;
        bool bret = m_pkey.Next(rid);
        if (!bret)
        {
            return NULL;
        }
        return &m_table.m_table[rid];
    }
};

class CPushThread : public CBF_Thread
{
public:
    CPushThread()
    {
        m_bIsRunning = false;
        m_nIndex = 0;
        m_nCurrDataNum = 1;
        m_nMaxDataNum = 9999999999999;
        m_TotalNs = 0;
        m_nBindCpu = 0;
    }
    virtual ~CPushThread()
    {
    }
    virtual int Run()
    {
        S_NOLOCK_DATA data;
        UINT64_ blocalus = 0;
        int blocalns = 0;
        UINT64_ elocalus = 0;
        int elocalns = 0;
        m_TotalNs = 0;
#ifndef _WINDOWS
        if (m_nBindCpu > 0)
        {
            cpu_set_t cpu_set;
            CPU_ZERO(&cpu_set);
            CPU_SET(m_nBindCpu + 1, &cpu_set);

            int ret = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set), &cpu_set);
            if (ret < 0)
            {
                printf("bind core fail\n");
            }
            else
            {
                printf("bind core success\n");
            }
        }
#endif
#ifdef _WINDOWS
        blocalus = CBF_Date_Time::GetTickCount();
        blocalns = 0;
#else
        CBF_Date_Time::GetTickNS(blocalus, blocalns);
#endif
        while (m_bIsRunning)
        {
            data.data_id = m_nIndex * 100000000000000 + m_nCurrDataNum;
#ifdef _WINDOWS
            data.push_us = CBF_Date_Time::GetTickCount();
            data.push_ns = 0;
#else
            // data.push_us = CBF_Date_Time::GetTickUS();
            CBF_Date_Time::GetTickNS(data.push_us, data.push_ns);
#endif
            if (g_QueueNoLock.Push(data))
            {
#ifdef _WINDOWS
                elocalus = CBF_Date_Time::GetTickCount();
                m_TotalNs += ((elocalus - data.push_us) * 1000000);
#else
                CBF_Date_Time::GetTickNS(elocalus, elocalns);
                m_TotalNs += ((elocalus - data.push_us) * 1000 + elocalns - data.push_ns);
#endif

#ifdef _WINDOWS
                if (g_nPushSleepTimeus / 1000 > 0)
                {
                    SLEEP(g_nPushSleepTimeus / 1000);
                }
#else
                if (g_nPushSleepTimeus > 0)
                {
                    usleep(g_nPushSleepTimeus);
                }
#endif
                m_nCurrDataNum++;
                if (m_nCurrDataNum > m_nMaxDataNum)
                {
#ifdef _WINDOWS
                    printf("push thread[%d] push %I64u success \n", m_nIndex, m_nCurrDataNum - 1);
#else
                    printf("\npush thread[%d] push %llu success \n", m_nIndex, m_nCurrDataNum - 1);
#endif
                    m_bIsRunning = false;
                    break;
                }
            }
        }
#ifdef _WINDOWS
        printf("push thread[%d] Stop %I64u次 success totaltime[%I64u]ms  平均push耗时[%.3lf]ns\n", m_nIndex, m_nCurrDataNum - 1, data.push_us - blocalus, 1.0 * m_TotalNs / (m_nCurrDataNum - 1));
#else
        printf("push thread[%d] Stop %llu次 success totaltime[%llu]us 平均push耗时[%.3lf]ns\n", m_nIndex, m_nCurrDataNum - 1, (data.push_us - blocalus), 1.0 * m_TotalNs / (m_nCurrDataNum - 1));
#endif
        return 0;
    }

    bool Start()
    {
        m_bIsRunning = true;
        CreateThread();
        return true;
    }
    void Stop()
    {
        m_bIsRunning = false;
        Join();
    }
    CBF_Date_Time m_pDate;
    bool m_bIsRunning;
    int m_nIndex;
    UINT64_ m_nMaxDataNum;
    UINT64_ m_nCurrDataNum;
    UINT64_ m_TotalNs;

    std::vector<UINT64_> m_vPopData;

    int m_nBindCpu;
};

class CPopThread : public CBF_Thread
{
public:
    CPopThread()
    {
        m_bIsRunning = false;
        m_nIndex = 0;
        m_nCurrDataNum = 0;
        m_bIsSave = false;
        m_nAddErrnum = 0;
        m_nBindCpu = 0;
        m_TotalNs = 0;
    }
    virtual ~CPopThread()
    {
    }
    virtual int Run()
    {
        S_NOLOCK_DATA data;
#ifndef _WINDOWS
        if (m_nBindCpu > 0)
        {
            cpu_set_t cpu_set;
            CPU_ZERO(&cpu_set);
            CPU_SET(m_nBindCpu + 1, &cpu_set);

            int ret = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set), &cpu_set);
            if (ret < 0)
            {
                printf("bind core fail\n");
            }
            else
            {
                printf("bind core success\n");
            }
        }
#endif
        UINT64_ pop_us = 0;
        int pop_ns = 0;
        INT64_ curns = 0;
        while (m_bIsRunning)
        {
            if (g_QueueNoLock.Pop(data))
            {
#ifdef _WINDOWS
                pop_us = CBF_Date_Time::GetTickCount();
                curns = (pop_us - data.push_us) * 1000000;
#else
                // pop_us = CBF_Date_Time::GetTickUS();
                // curns = (pop_us- data.push_us)*1000;
                CBF_Date_Time::GetTickNS(pop_us, pop_ns);
                curns = (pop_us - data.push_us) * 1000 + pop_ns - data.push_ns;
                // printf("data_id[%llu] pop-push耗时[%lld]ns\n",data.data_id,curns);
#endif
                m_TotalNs += curns;
                m_nCurrDataNum++;
                if (!m_popdata.Add(data))
                {
                    m_nAddErrnum++;
                }
            }
            else
            {
                if (1 == g_nExitPush)
                {
                    break;
                }
            }
        }
#ifdef _WINDOWS
        printf("pop thread[%d] Stop %I64u success 平均push-pop耗时[%.3lf]ns\n", m_nIndex, m_nCurrDataNum, 1.0 * m_TotalNs / m_nCurrDataNum);
#else
        printf("pop thread[%d] Stop %llu success 平均push-pop耗时[%.3lf]ns\n", m_nIndex, m_nCurrDataNum, 1.0 * m_TotalNs / m_nCurrDataNum);
#endif
        return 0;
    }

    bool Start()
    {
        m_bIsRunning = true;
        CreateThread();
        return true;
    }
    void Stop()
    {
        m_bIsRunning = false;
        Join();
    }
    bool m_bIsSave;
    bool m_bIsRunning;
    int m_nIndex;
    UINT64_ m_nCurrDataNum;
    CTbl_PopData m_popdata;
    int m_nAddErrnum;
    int m_nBindCpu;

    UINT64_ m_TotalNs; // pop出来计算的每笔push pop的总耗时
};

CPopThread poparray[20];
CPushThread pusharray[20];

bool RunOne(int pushnum, int popnum, int nPushNumPerThread, int cpucore = 0, int testdelay = 0)
{

    int i = 0;
    for (i = 0; i < 20; i++)
    {
        poparray[i].m_nIndex = i;
        poparray[i].m_nCurrDataNum = 0;
        poparray[i].m_TotalNs = 0;
        poparray[i].m_popdata.Clear();

        pusharray[i].m_nIndex = i;
        pusharray[i].m_nCurrDataNum = 1;
        pusharray[i].m_TotalNs = 0;
        pusharray[i].m_nMaxDataNum = nPushNumPerThread;
    }
    if (testdelay == 1) // 测试延时，要先启pop线程等待
    {
        g_nExitPush = 0;
        for (i = 0; i < popnum; i++)
        {
            if (cpucore > 0)
            {
                poparray[i].m_nBindCpu = cpucore + i;
            }
            poparray[i].Start();
        }
        for (i = 0; i < pushnum; i++)
        {
            if (cpucore > 0)
            {
                pusharray[i].m_nBindCpu = cpucore + popnum + i;
            }
            pusharray[i].Start();
        }
        // for (i = 0; i < pushnum; i++)
        //{
        //     //等待push线程退出
        //     pusharray[i].Join();
        // }
        while (true)
        {
            int startnum = pushnum;
            for (i = 0; i < pushnum; i++)
            {
                if (pusharray[i].IsStoped())
                {
                    startnum--;
                }
            }
            if (startnum < 1)
            {
                break;
            }
            SLEEP(100);
        }
        SLEEP_SECONDS(2);
        for (i = 0; i < popnum; i++)
        {
            // 等待pop线程退出
            poparray[i].Stop();
        }
    }
    else
    {
        g_nExitPush = 0;
        for (i = 0; i < pushnum; i++)
        {
            if (cpucore > 0)
            {
                pusharray[i].m_nBindCpu = cpucore + popnum + i;
            }
            pusharray[i].Start();
        }
        for (i = 0; i < popnum; i++)
        {
            if (cpucore > 0)
            {
                poparray[i].m_nBindCpu = cpucore + i;
            }
            poparray[i].Start();
        }
        // for (i = 0; i < pushnum; i++)
        //{
        //     //等待push线程退出
        //     pusharray[i].Join();
        // }
        // 等待全部push线程退出
        while (true)
        {
            int startnum = pushnum;
            for (i = 0; i < pushnum; i++)
            {
                if (pusharray[i].IsStoped())
                {
                    startnum--;
                }
            }
            if (startnum < 1)
            {
                break;
            }
            SLEEP(100);
        }
        g_nExitPush = 1;
        while (true)
        {
            int startnum = popnum;
            for (i = 0; i < popnum; i++)
            {
                if (poparray[i].IsStoped())
                {
                    startnum--;
                }
            }
            if (startnum < 1)
            {
                break;
            }
            SLEEP(100);
        }
    }

    UINT64_ nTotalPushNum = 0;
    UINT64_ nTotalPopNum = 0;
    UINT64_ nTotalTimens = 0;
    UINT64_ nTotalPushPopns = 0;
    for (i = 0; i < pushnum; i++)
    {
        nTotalPushNum += (pusharray[i].m_nCurrDataNum - 1);
        nTotalTimens += pusharray->m_TotalNs;
    }
    for (i = 0; i < popnum; i++)
    {
        nTotalPopNum += poparray[i].m_nCurrDataNum;
        nTotalPushPopns += poparray[i].m_TotalNs;
    }
#ifdef _WINDOWS
    printf("push队列总数[%I64u] push总耗时[%I64u]ns 平均每笔耗时[%.3lf]ns pop总数[%I64u] 平均push-pop耗时[%.3lf]ns\n", nTotalPushNum, nTotalTimens, 1.0 * nTotalTimens / nTotalPopNum, nTotalPopNum, 1.0 * nTotalPushPopns / nTotalPopNum);
#else
    printf("push队列总数[%llu] push总耗时[%llu]ns 平均每笔耗时[%.3lf]ns pop总数[%llu] 平均push-pop耗时[%.3lf]ns\n", nTotalPushNum, nTotalTimens, 1.0 * nTotalTimens / nTotalPopNum, nTotalPopNum, 1.0 * nTotalPushPopns / nTotalPopNum);
#endif

    int err = 0;
    int j = 0;
    int k = 0;
    UINT64_ data_id = 0;
    int findnum = 0;
    for (i = 0; i < pushnum; i++)
    {
        for (j = 1; j < pusharray[i].m_nCurrDataNum; j++)
        {
            data_id = i * 100000000000000 + j;
            // 从pop线程里查找
            findnum = 0;
            for (k = 0; k < popnum; k++)
            {
                if (poparray[k].m_popdata.Select(data_id))
                {
                    findnum++;
                }
            }
            if (findnum != 1)
            {
                err++;
            }
        }
    }
    if (err == 0)
    {
        printf("push/pop数据校验通过\n");
        return true;
    }
    else
    {
        printf("push/pop 数据校验不通过 \n");
        return false;
    }
}

int main(int argc, char *argv[])
{
    std::cout << "Hello World!\n"
              << std::endl;

    UINT64_ nTotalLoopnum = 1;
    printf("测试无锁队列\n");
    printf("测试无锁队列 cmd pushnum popnum datanum cpucore loopnum testdelay pushsleepus\n");
    char tmpcahr[200];
    bzero(tmpcahr, sizeof(tmpcahr));
    int nPushThread;
    int nPopThread;
    UINT64_ nPushNumPerThread = 0;
    int beginbindcpu = 0;
    int testdelay = 0;
    int i = 0;
    g_nPushSleepTimeus = 0;

    if (argc == 8)
    {
        nPushThread = atoi(argv[1]);
        if (nPushThread < 1 || nPushThread > 20)
        {
            printf("push数据个数有误\n");
            return -1;
        }
        nPopThread = atoi(argv[2]);
        if (nPopThread < 1 || nPopThread > 20)
        {
            printf("pop线程输入有误\n");
            return -1;
        }
        nPushNumPerThread = atoi(argv[3]);
        if (nPushNumPerThread < 1)
        {
            printf("push数据个数有误\n");
            return -1;
        }
        if (nPushNumPerThread >= 100000000000000)
        {
            printf("push数据个数有误\n");
            return -1;
        }
        beginbindcpu = atoi(argv[4]);
        nTotalLoopnum = atoi(argv[5]);
        testdelay = atoi(argv[6]);
        g_nPushSleepTimeus = atoi(argv[7]);
    }
    else
    {
        CBF_Tools::GetConsoleEnter("请输入push线程数 1-20", 2, tmpcahr);
        nPushThread = atoi(tmpcahr);
        if (nPushThread < 1 || nPushThread >= 20)
        {
            printf("push线程输入有误\n");
            return -1;
        }
        CBF_Tools::GetConsoleEnter("请输入pop线程数 1-20", 2, tmpcahr);
        nPopThread = atoi(tmpcahr);
        if (nPopThread < 1 || nPopThread > 20)
        {
            printf("pop线程输入有误\n");
            return -1;
        }
        CBF_Tools::GetConsoleEnter("请输入每个线程push数据个数", 15, tmpcahr);
#ifdef _WINDOWS
        nPushNumPerThread = atoll(tmpcahr);
#else
        nPushNumPerThread = atoll(tmpcahr);
#endif
        if (nPushNumPerThread < 1)
        {
            printf("push数据个数有误\n");
            return -1;
        }
        if (nPushNumPerThread >= 100000000000000)
        {
            printf("push数据个数有误\n");
            return -1;
        }
        CBF_Tools::GetConsoleEnter("请输入循环个数", 10, tmpcahr);
        nTotalLoopnum = atoi(tmpcahr);
        CBF_Tools::GetConsoleEnter("请输入是否测试延时数据0不测1测", 1, tmpcahr);
        testdelay = atoi(tmpcahr);
        CBF_Tools::GetConsoleEnter("请输入push线程sleep时间us，小于等0不sleep", 1, tmpcahr);
        g_nPushSleepTimeus = atoi(tmpcahr);
    }
    for (i = 0; i < nTotalLoopnum; i++)
    {
        printf("第[%d]次循环测试  PushThread[%d] PopThread[%d] PushNumPerThread[%d] bindcpu[%d]\n", i + 1, nPushThread, nPopThread, nPushNumPerThread, beginbindcpu);
        UINT64_ btime = CBF_Date_Time::GetTickUS();
        if (!RunOne(nPushThread, nPopThread, nPushNumPerThread, beginbindcpu))
        {
            return -1;
        }
        UINT64_ etime = CBF_Date_Time::GetTickUS();
#ifdef _WINDOWS
        printf("第[%d]次循环测试 总耗时[%I64u]us\n", i + 1, etime - btime);
#else
        printf("第[%d]次循环测试 总耗时[%llu]us\n", i + 1, etime - btime);
#endif
    }

    return 0;
}

// 运行程序: Ctrl + F5 或调试 >“开始执行(不调试)”菜单
// 调试程序: F5 或调试 >“开始调试”菜单

// 入门使用技巧:
//   1. 使用解决方案资源管理器窗口添加/管理文件
//   2. 使用团队资源管理器窗口连接到源代码管理
//   3. 使用输出窗口查看生成输出和其他消息
//   4. 使用错误列表窗口查看错误
//   5. 转到“项目”>“添加新项”以创建新的代码文件，或转到“项目”>“添加现有项”以将现有代码文件添加到项目
//   6. 将来，若要再次打开此项目，请转到“文件”>“打开”>“项目”并选择 .sln 文件
